import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;


/**
 * @author wangzj
 * @description Flink状态的练习
 * 每当第一个元素的和达到二，就把第二个元素的和与第一个元素的和相除，最后输出
 * @date 2020/7/23 23:48
 */
public class KeyedStateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(1L, 5L), Tuple2.of(1L, 2L))
                .keyBy(0)
                .flatMap(new CountWindowAverage())
                .printToErr();
        env.execute("KeyedStateDemo");
    }

    /**
     * 每当第一个元素的和达到二，就把第二个元素的和与第一个元素的和相除
     */
    public static class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {

        //存储历史的求和数据
        private transient ValueState<Tuple2<Long, Long>> sum;

        @Override
        public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Long>> out) throws Exception {
            Tuple2<Long, Long> currentSum;
            // 访问ValueState。value方法获取的就是sum的状态值，第一次获取的是null
            if (sum.value() == null) {
                currentSum = Tuple2.of(0L, 0L);
            } else {
                currentSum = sum.value();
            }
            //第一个元素相加
            currentSum.f0 += value.f0;
            //第二个元素累加
            currentSum.f1 += value.f1;
            // 更新state
            sum.update(currentSum);
            // 如果第一个元素的值大于等于2，求结果并清空state
            if (currentSum.f0 >= 2) {
                //输出结果
                out.collect(new Tuple2<>(value.f0, currentSum.f1 / currentSum.f0));
                sum.clear();
            }
        }

        //TODO 不是很理解下面代码的意思
        public void open(Configuration config) {
            ValueStateDescriptor<Tuple2<Long, Long>> descriptor = new ValueStateDescriptor(
                    "average", // state的名字
                    TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {
                    })
            );

            //设置它们的过期时间
            StateTtlConfig ttlConfig = StateTtlConfig
                    .newBuilder(Time.seconds(10))
                    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                    .build();
            descriptor.enableTimeToLive(ttlConfig);

            //来获取状态的句柄
            sum = getRuntimeContext().getState(descriptor);
        }
    }
}


